package com.binance.notification.web.service.ws;

import java.math.BigDecimal;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.binance.assetservice.vo.response.product.FetchProductsResponse;
import com.binance.distcache.distlock.DistLockManager;
import com.binance.distcache.distlock.IDistLock;
import com.binance.notification.web.kafka.Productor;
import com.binance.notification.web.service.helper.ProductService;
import com.binance.notification.web.utils.Constants;
import com.binance.notification.web.utils.Worker;
import com.github.rholder.retry.Attempt;
import com.github.rholder.retry.RetryListener;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import com.google.common.collect.Maps;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class ProductRealtimePriceService {
    private volatile WebSocketClient client;
    private volatile boolean stop;
    @Resource
    private Productor productor;
    @Value("${matchbox.ws.connection.lost.timeout:60}")
    private int connectionLostTimeout;
    @Value("${matchbox.ws.trade.price.url}")
    private String wsUrl;
    @Autowired
    private DistLockManager distLockManager;
    @Autowired
    private ProductService productService;
    @Autowired
    private Worker worker;
    @Value("${jpush.alert.price.enabled:false}")
    private boolean alertPriceEnabled;
    private Map<String, Product> productCache = Maps.newConcurrentMap();
    private volatile boolean locked;

    // 用来获取并缓存所有产品，直到成功获取并缓存，否则将每隔30秒中执行一次
    private final Retryer<Void> retryer = RetryerBuilder.<Void>newBuilder().retryIfExceptionOfType(Throwable.class)
            .withRetryListener(new RetryListener() {

                @Override
                public <V> void onRetry(Attempt<V> attempt) {
                    if (attempt.hasException()) {
                        System.out.println(System.currentTimeMillis());
                        log.warn(String.format("%d retry after %d ms", attempt.getAttemptNumber(),
                                attempt.getDelaySinceFirstAttempt()), attempt.getExceptionCause());
                    }
                }

            }).withWaitStrategy(WaitStrategies.fixedWait(30, TimeUnit.SECONDS))
            .withStopStrategy(StopStrategies.neverStop()).build();

    @PostConstruct
    public void init() {
        this.initProductCache();
        this.start();

        IDistLock lock = distLockManager.buildRedDistLock(Constants.NOTIFICATION_WEBSOCKET_LOCK);
        worker.scheduleAtFixedRate(() -> {
            if (!locked) {
                if (locked = lock.tryLock(10, -1, TimeUnit.SECONDS)) {
                    // 获取到锁
                    // log.info("节点{}获取到redis锁成为与撮合系统通信的主服务器", System.getProperty("local-ip"));
                    // this.start();
                } else {
                    // 没有获取到锁，需要将websocket停止
                    locked = false;
                    // log.info("节点{}没有获取到redis锁", System.getProperty("local-ip"));
                    // this.stop();
                }
            }
            return null;
        }, 0, Constants.REDIS_WX_LOCK_SCHEDULED * 1000);
    }

    public void initProductCache() {
        worker.submit(() -> {
            try {
                retryer.call(() -> {
                    for (FetchProductsResponse resp : this.productService.getAllProducts()) {
                        this.productCache.put(resp.getSymbol(), new Product(resp.getQuoteAsset(), resp.getClose(),
                                resp.getTickSize().stripTrailingZeros().scale()));
                    }
                    return null;
                });
            } catch (Exception e) {
            }
        });
    }

    private void initWebSocket() {
        if (client == null) {
            client = new WebSocketClient(URI.create(wsUrl)) {

                @Override
                public void onOpen(ServerHandshake arg0) {
                    log.info("打开{}的WebSocket链接：{}", wsUrl, arg0);
                }

                @Override
                public void onMessage(String arg0) {
                    if (!stop) {
                        handlerReceiveMessage(arg0);
                    }
                }

                @Override
                public void onError(Exception arg0) {
                    log.error("WebSocket发生错误：", arg0);
                }

                @Override
                public void onClose(int arg0, String arg1, boolean arg2) {
                    log.info("链接已关闭：{},{},{}", arg0, arg1, arg2);
                    if (!stop) {
                        restart();
                    }
                }

                @Override
                public void onMessage(ByteBuffer bytes) {}

            };
            client.setConnectionLostTimeout(connectionLostTimeout);
        }
    }

    private void start() {
        this.stop = false;
        initWebSocket();
        if (client.isConnecting() || client.isOpen()) {
            return;
        }
        client.connect();
    }

    private void restart() {
        log.info("1秒后重新对{}地址发起WebSocket连接......", wsUrl);
        worker.schedule(() -> {
            initWebSocket();
            if (client.isClosed() || client.isClosing()) {
                client.reconnect();
            }
        }, 1000);
    }

    private void stop() {
        this.stop = true;
        if (client != null) {
            this.client.close();
        }
    }

    private void handlerReceiveMessage(String message) {
        log.info("WebSocket收到的交易对的最新价格的消息：{}", message);
        JSONArray array = JSON.parseArray(message);
        for (int i = 0; i < array.size(); i++) {
            JSONObject object = (JSONObject) array.get(i);
            String symbol = object.getString("s");
            Double close = object.getDouble("c");
            Product p = this.productCache.get(symbol);
            if (p != null) {
                p.setClose(BigDecimal.valueOf(close));
            }
            if (locked && this.alertPriceEnabled) {
                String payload = symbol + "|" + close;
                productor.sendAlertPricePairMessage(payload);
            }
        }
    }

    // public BigDecimal getCny(String symbol) {
    // BigDecimal usd = this.getUsd(symbol);
    // return usd.multiply(new BigDecimal(this.productService.getCnyUsd()));
    // }

    public BigDecimal getUsd(String symbol) {
        Product p = this.productCache.get(symbol);
        if (p == null) {
            return BigDecimal.ZERO;
        }
        if ("USDT".equalsIgnoreCase(p.getQuoteAsset())) {
            return p.getClose();
        }
        Product p2 = this.productCache.get(p.getQuoteAsset() + "USDT");
        if (p2 == null) {
            return BigDecimal.ZERO;
        }
        return p.getClose().multiply(p2.getClose()).setScale(p.getTickSize(), BigDecimal.ROUND_CEILING);
    }

    @Getter
    @Setter
    private static class Product {
        private BigDecimal close;
        private int tickSize;
        private String quoteAsset;

        Product(String quoteAsset, BigDecimal close, int tickSize) {
            this.tickSize = tickSize;
            this.close = close;
            this.quoteAsset = quoteAsset;
        }

        public BigDecimal getClose() {
            if (this.close == null) {
                return BigDecimal.ZERO;
            }
            return this.close;
        }
    }
}
